package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggest;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.AlbumIndexRepository;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;
import java.util.stream.Stream;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {

    @Resource
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Resource
    private CategoryFeignClient categoryFeignClient;

    @Resource
    private UserInfoFeignClient userInfoFeignClient;

    @Resource
    private AlbumIndexRepository albumIndexRepository;

    @Resource
    private ThreadPoolExecutor threadPoolExecutor;

    @Resource
    private ElasticsearchClient elasticsearchClient;

    @Resource
    private SuggestIndexRepository suggestIndexRepository;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void upperAlbum(Long albumId) {
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        //远程调用：根据专辑id获取到专辑的值
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
            //判断专辑不为空
            Assert.notNull(albumInfoResult, "获取到的专辑信息为空");
            AlbumInfo albumInfo = albumInfoResult.getData();
            Assert.notNull(albumInfo, "专辑对象为空");
            //属性拷贝到albumInfoIndex中
            BeanUtils.copyProperties(albumInfo, albumInfoIndex);
            return albumInfo;
        }, threadPoolExecutor);

        //远程调用获取到用户的值
        CompletableFuture<Void> userInfoCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfoVo(albumInfo.getUserId());
            Assert.notNull(userInfoVoResult, "获取到的用户信息为空");
            UserInfoVo userInfoVo = userInfoVoResult.getData();
            Assert.notNull(userInfoVo, "用户对象为空");
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        }, threadPoolExecutor);
        //远程调用：根据三级分类id获取的分类数据
        CompletableFuture<Void> categoryCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<BaseCategoryView> categoryViewResult = categoryFeignClient.getCategoryView(albumInfo.getCategory3Id());
            Assert.notNull(categoryViewResult, "获取到的分类信息为空");
            //给albumInfoIndex赋值
            BaseCategoryView baseCategoryView = categoryViewResult.getData();
            Assert.notNull(baseCategoryView, "分类对象为空");
            albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
            albumInfoIndex.setCategory3Id(baseCategoryView.getCategory3Id());
        }, threadPoolExecutor);
        //设置专辑热度
        //更新统计量与得分，默认随机，方便测试
        int num1 = new Random().nextInt(1000);
        int num2 = new Random().nextInt(100);
        int num3 = new Random().nextInt(50);
        int num4 = new Random().nextInt(300);
        albumInfoIndex.setPlayStatNum(num1);
        albumInfoIndex.setSubscribeStatNum(num2);
        albumInfoIndex.setBuyStatNum(num3);
        albumInfoIndex.setCommentStatNum(num4);
        double hotScore = num1 * 0.2 + num2 * 0.3 + num3 * 0.4 + num4 * 0.1;
        //  设置热度排名
        albumInfoIndex.setHotScore(hotScore);
        //获取专辑属性值
        CompletableFuture<Void> AttributeCompletableFuture = CompletableFuture.runAsync(() -> {
            Result<List<AlbumAttributeValue>> albumAttributeValue = albumInfoFeignClient.findAlbumAttributeValue(albumId);
            Assert.notNull(albumAttributeValue, "获取到的专辑属性信息为空");
            List<AlbumAttributeValue> albumAttributeValues = albumAttributeValue.getData();
            if (!CollectionUtils.isEmpty(albumAttributeValues)) {
                List<AttributeValueIndex> attributeValueIndices = albumAttributeValues.stream().map(item -> {
                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                    BeanUtils.copyProperties(item, attributeValueIndex);
                    return attributeValueIndex;
                }).toList();
                albumInfoIndex.setAttributeValueIndexList(attributeValueIndices);
            }
        }, threadPoolExecutor);
        CompletableFuture.allOf(albumInfoCompletableFuture
                , userInfoCompletableFuture
                , AttributeCompletableFuture
                , categoryCompletableFuture).join();
        //保存数据
        albumIndexRepository.save(albumInfoIndex);

        //初始化专辑提词
        SuggestIndex suggestAlbumTitleIndex = new SuggestIndex();
        suggestAlbumTitleIndex.setId(albumId.toString());
        suggestAlbumTitleIndex.setTitle(albumInfoIndex.getAlbumTitle());
        suggestAlbumTitleIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()}));
        suggestAlbumTitleIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));
        suggestAlbumTitleIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));
        suggestIndexRepository.save(suggestAlbumTitleIndex);

        //  专辑简介提词
        SuggestIndex albumIntroSuggestIndex = new SuggestIndex();
        albumIntroSuggestIndex.setId(UUID.randomUUID().toString().replaceAll("-", ""));
        albumIntroSuggestIndex.setTitle(albumInfoIndex.getAlbumIntro());
        albumIntroSuggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumIntro()}));
        albumIntroSuggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumIntro())}));
        albumIntroSuggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumIntro())}));
        this.suggestIndexRepository.save(albumIntroSuggestIndex);

        // 专辑主播提词
        SuggestIndex announcerSuggestIndex = new SuggestIndex();
        announcerSuggestIndex.setId(UUID.randomUUID().toString().replaceAll("-", ""));
        announcerSuggestIndex.setTitle(albumInfoIndex.getAnnouncerName());
        announcerSuggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAnnouncerName()}));
        announcerSuggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAnnouncerName())}));
        announcerSuggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAnnouncerName())}));
        suggestIndexRepository.save(announcerSuggestIndex);
    }

    @Override
    public void lowerAlbum(Long albumId) {
        albumIndexRepository.deleteById(albumId);
    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) {
        try {
            SearchRequest searchRequest = this.queryDsl(albumIndexQuery);
            SearchResponse<AlbumInfoIndex> result = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
            AlbumSearchResponseVo albumSearchResponseVo = this.parseResultData(result);
            albumSearchResponseVo.setPageNo(albumIndexQuery.getPageNo());
            albumSearchResponseVo.setPageSize(albumIndexQuery.getPageSize());
            Long totalPages = (albumSearchResponseVo.getTotal() + albumIndexQuery.getPageSize() - 1) / albumIndexQuery.getPageSize();
            albumSearchResponseVo.setTotalPages(totalPages);
            return albumSearchResponseVo;
        } catch (IOException e) {
            throw new GuiguException(ResultCodeEnum.FAIL);
        }
    }

    @Override
    public List<Map> channel(Long category1Id) {
        //远程调用：根据一级分类id获取到对应的三级分类数据
        Result<List<BaseCategory3>> result = categoryFeignClient.findTopBaseCategory3(category1Id);
        Assert.notNull(result, "返回的三级分类消息为空");
        List<BaseCategory3> baseCategory3List = result.getData();
        //将这个集合转换为map集合方便后面取到三级分类数据
        Map<Long, BaseCategory3> category3Map = baseCategory3List.stream()
                .collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));
        Assert.notNull(baseCategory3List, "返回的三级分类数据为空");
        //通过三级分类集合得到三级分类id集合
        List<Long> baseCategory3IdList = baseCategory3List.stream().map(BaseCategory3::getId).toList();
        List<FieldValue> fieldValues = baseCategory3IdList.stream().map(FieldValue::of).toList();
        //生成dsl语句
        try {
            SearchResponse<AlbumInfoIndex> response = elasticsearchClient
                    .search(s -> s.index("albuminfo")
                                    .query(q -> q.terms(t -> t.field("category3Id").terms(f ->
                                            f.value(fieldValues))))
                                    .aggregations("category3IdAgg", a -> a.terms(t ->
                                                    t.field("category3Id")
                                                            .size(10))
                                            .aggregations("topAgg", f -> f.topHits(t ->
                                                    t.size(6)
                                                            .sort(so -> so.field(f1 -> f1
                                                                    .field("buyStatNum")
                                                                    .order(SortOrder.Desc)))
                                            ))
                                    )
                            , AlbumInfoIndex.class);
            //获取聚合的结果集
            Aggregate category3IdAgg = response.aggregations().get("category3IdAgg");
            List<Map> list = category3IdAgg.lterms().buckets().array().stream().map(item -> {
                //获取三级分类id
                long category3Id = item.key();
                //获取topAgg
                Aggregate topAgg = item.aggregations().get("topAgg");
                List<Hit<JsonData>> hits = topAgg.topHits().hits().hits();
                List<AlbumInfoIndex> albumInfoIndices = hits.stream().map(hitJson -> {
                    //获取到数据
                    String json = hitJson.source().toString();
                    AlbumInfoIndex albumInfoIndex = JSON.parseObject(json, AlbumInfoIndex.class);
                    return albumInfoIndex;
                }).toList();
                Map map = new HashMap<>();
                //  存储根据三级分类Id要找到的三级分类
                map.put("baseCategory3", category3Map.get(category3Id));
                //  存储所有的专辑集合数据
                map.put("list", albumInfoIndices);
                return map;
            }).toList();
            return list;
        } catch (IOException e) {
            throw new GuiguException(ResultCodeEnum.FAIL);
        }
    }

    @Override
    public List<String> completeSuggest(String keyword) {
        //编写dsl语句
        try {
            //执行dsl语句
            SearchResponse<SuggestIndex> response = elasticsearchClient.search(s -> s.index("suggestinfo")
                            .suggest(su -> su.suggesters("suggestionKeyword", f -> f
                                            .prefix(keyword)
                                            .completion(c -> c
                                                    .field("keyword")
                                                    .skipDuplicates(true)
                                                    .fuzzy(z -> z.fuzziness("auto"))
                                            )
                                    ).suggesters("suggestionkeywordPinyin", f -> f
                                            .prefix(keyword)
                                            .completion(c -> c
                                                    .field("keywordPinyin")
                                                    .skipDuplicates(true)
                                                    .fuzzy(z -> z.fuzziness("auto"))
                                            )
                                    ).suggesters("suggestionkeywordSequence", f -> f
                                            .prefix(keyword)
                                            .completion(c -> c
                                                    .field("keywordSequence")
                                                    .skipDuplicates(true)
                                                    .fuzzy(z -> z.fuzziness("auto"))
                                            )
                                    )
                            )
                            .size(10)
                            .sort(o -> o.field(f -> f.field("_score").order(SortOrder.Desc)))
                    , SuggestIndex.class);
            //获取结果
            Map<String, List<Suggestion<SuggestIndex>>> suggest = response.suggest();
            HashSet<String> titleSet = new HashSet<>();
            titleSet.addAll(this.parseResultData(response, "suggestionKeyword"));
            titleSet.addAll(this.parseResultData(response, "suggestionkeywordPinyin"));
            titleSet.addAll(this.parseResultData(response, "suggestionkeywordSequence"));
            //判断获取到的结果有没有10条
            if (titleSet.size() < 10) {
                //  使用查询数据的方式来填充集合数据，让这个提示信息够10条数据.
                SearchResponse<SuggestIndex> response1 = null;
                try {
                    response1 = elasticsearchClient.search(s -> s.index("suggestinfo")
                                    .query(f -> f.match(m -> m.field("title").query(keyword)))
                            , SuggestIndex.class);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                //  从查询结果集中获取数据
                for (Hit<SuggestIndex> hit : response.hits().hits()) {
                    //  获取数据结果
                    SuggestIndex suggestIndex = hit.source();
                    //  获取titile
                    titleSet.add(suggestIndex.getTitle());
                    //  判断当前这个结合的长度.
                    if (titleSet.size() == 10) {
                        break;
                    }
                }
            }
            return new ArrayList<>(titleSet);
        } catch (IOException e) {
            throw new GuiguException(ResultCodeEnum.FAIL);
        }
    }

    @Override
    public void updateLatelyAlbumRanking() {
        //先查询所有的一级分类id
        Result<List<BaseCategory1>> baseCategory1Result = categoryFeignClient.findAllCategory1();
        //  获取集合
        Assert.notNull(baseCategory1Result, "一级分类结果集为空");
        List<BaseCategory1> baseCategory1List = baseCategory1Result.getData();
        Assert.notNull(baseCategory1List, "一级分类集合为空");
        //构建排行榜的分类
        String[] rankingDimensionArray = new String[]{"hotScore", "playStatNum",
                "subscribeStatNum", "buyStatNum", "commentStatNum"};
        //循环遍历集合，查询es得到数据
        for (BaseCategory1 baseCategory1 : baseCategory1List) {
            for (String ranging : rankingDimensionArray) {
                try {
                    SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(s -> s.index("albuminfo")
                                    .query(q -> q.term(t -> t.field("category1Id").value(baseCategory1.getId())))
                                    .sort(f -> f.field(f1 -> f1.field(ranging).order(SortOrder.Desc)))
                                    .size(10)
                            , AlbumInfoIndex.class);
                    //  获取到当前执行结果集，将数据存储到缓存：
                    List<AlbumInfoIndex> albumInfoIndexList = response.hits().hits()
                            .stream().map(Hit::source).collect(Collectors.toList());
                    //向redis里添加缓存
                    String key = RedisConstant.RANKING_KEY_PREFIX + baseCategory1.getId();
                    stringRedisTemplate.opsForHash().put(key, ranging, JSON.toJSONString(albumInfoIndexList));
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String dimension) {
        String key = RedisConstant.RANKING_KEY_PREFIX + +category1Id;
        String jsonStr = (String) stringRedisTemplate.opsForHash().get(key, dimension);
        return JSON.parseObject(jsonStr, List.class);
    }

    private List<String> parseResultData(SearchResponse<SuggestIndex> response, String keyword) {
        //获取到结果集
        List<Suggestion<SuggestIndex>> suggestions = response.suggest().get(keyword);
        ArrayList<String> list = new ArrayList<>();
        //判断不为空
        if (!CollectionUtils.isEmpty(suggestions)) {
            suggestions.forEach(item -> {
                CompletionSuggest<SuggestIndex> completion = item.completion();
                List<String> strings = completion.options().stream().map(c -> {
                    String title = c.source().getTitle();
                    return title;
                }).toList();
                list.addAll(strings);
            });
        }
        return list;
    }

    /**
     * 将查询到的数据转换为要的数据
     *
     * @param result
     * @return
     */
    private AlbumSearchResponseVo parseResultData(SearchResponse<AlbumInfoIndex> result) {
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        List<Hit<AlbumInfoIndex>> hits = result.hits().hits();
        if (!CollectionUtils.isEmpty(hits)) {
            List<AlbumInfoIndexVo> albumInfoIndexVos = hits.stream().map(item -> {
                AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                AlbumInfoIndex source = item.source();
                BeanUtils.copyProperties(source, albumInfoIndexVo);
                //  判断用户是否根据关键词进行检索.
                if (null != item.highlight().get("albumTitle")) {
                    //  获取高亮数据
                    String albumTitle = item.highlight().get("albumTitle").get(0);
                    //  赋值高亮数据
                    albumInfoIndexVo.setAlbumTitle(albumTitle);
                }
                return albumInfoIndexVo;
            }).toList();
            albumSearchResponseVo.setList(albumInfoIndexVos);
        }
        albumSearchResponseVo.setTotal(result.hits().total().value());
        return albumSearchResponseVo;
    }

    /**
     * 获取查询的dsl语句
     *
     * @param albumIndexQuery
     * @return
     */
    private SearchRequest queryDsl(AlbumIndexQuery albumIndexQuery) {
        //构造SearchRequest.Builder对象，通过它得到SearchRequest对象
        SearchRequest.Builder searchBuilder = new SearchRequest.Builder();
        searchBuilder.index("albuminfo");
        //构造bool对象
        BoolQuery.Builder boolBuilder = new BoolQuery.Builder();
        //判断关键词是否为空,如果不为空构造查询参数
        if (StringUtils.hasText(albumIndexQuery.getKeyword())) {
            //构造查询参数
            boolBuilder.should(s -> s.match(m -> m.field("albumTitle").query(albumIndexQuery.getKeyword())));
            boolBuilder.should(s -> s.match(m -> m.field("albumIntro").query(albumIndexQuery.getKeyword())));
            boolBuilder.should(s -> s.match(m -> m.field("announcerName").query(albumIndexQuery.getKeyword())));
            //对关键字设置高亮
            searchBuilder.highlight(h -> h.fields("albumTitle", f ->
                    f.preTags("<span style=color:#f86442>")
                            .postTags("</span>")));
        }
        //判断是否有一级分类id
        if (albumIndexQuery.getCategory1Id() != null && StringUtils.hasText(String.valueOf(albumIndexQuery.getCategory1Id()))) {
            boolBuilder.filter(f -> f.term(t -> t.field("category1Id").value(albumIndexQuery.getCategory1Id())));
        }
        //判断是否有二级分类id
        if (albumIndexQuery.getCategory2Id() != null && StringUtils.hasText(String.valueOf(albumIndexQuery.getCategory2Id()))) {
            boolBuilder.filter(f -> f.term(t -> t.field("category2Id").value(albumIndexQuery.getCategory2Id())));
        }
        //判断是否有三级分类id
        if (albumIndexQuery.getCategory3Id() != null && StringUtils.hasText(String.valueOf(albumIndexQuery.getCategory3Id()))) {
            boolBuilder.filter(f -> f.term(t -> t.field("category3Id").value(albumIndexQuery.getCategory3Id())));
        }
        //判断属性集合是否为空
        List<String> attributeList = albumIndexQuery.getAttributeList();
        if (!CollectionUtils.isEmpty(attributeList)) {
            for (String attribute : attributeList) {
                String[] split = attribute.split(":");
                if (split != null && split.length == 2) {
                    boolBuilder.filter(f ->
                            f.nested(n ->
                                    n.path("attributeValueIndexList")
                                            .query(q ->
                                                    q.bool(
                                                            b -> b.must(
                                                                    m -> m.term(
                                                                            t -> t.field("attributeValueIndexList.attributeId")
                                                                                    .value(split[0])
                                                                    )
                                                            ).must(
                                                                    m -> m.term(
                                                                            t -> t.field("attributeValueIndexList.valueId")
                                                                                    .value(split[1])
                                                                    )
                                                            )
                                                    )
                                            )
                            )
                    );
                }
            }
        }

        //设置分页
        searchBuilder.from((albumIndexQuery.getPageNo() - 1) * albumIndexQuery.getPageSize());
        searchBuilder.size(albumIndexQuery.getPageSize());

        //排序
        String order = albumIndexQuery.getOrder();
        if (StringUtils.hasText(order)) {
            String[] split = order.split(":");
            if (split != null && split.length == 2) {
                String filed = "";
                switch (split[0]) {
                    case "1":
                        filed = "hotScore";
                        break;
                    case "2":
                        filed = "playStatNum";
                        break;
                    case "3":
                        filed = "createTime";
                        break;
                    default:
                        break;
                }
                String finalFiled = filed;
                searchBuilder.sort(s ->
                        s.field(f -> f.field(finalFiled).order("asc".equals(split[1]) ? SortOrder.Asc : SortOrder.Desc))
                );
            }
        }
        searchBuilder.query(boolBuilder.build()._toQuery());

        return searchBuilder.build();
    }
}
